import os
from _ast import List
from functools import reduce
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, isnan  # 使用 Spark 的 isnan
from SparkSessionBase import SparkSessionBase  # 假设这是你自己的封装类

# 继承 SparkSessionBase 类
class TextRandJob(SparkSessionBase):
    SPARK_URL = "yarn"
    SPARK_APP_NAME = 'clean_checkin'
    ENABLE_HIVE_SUPPORT = True

    def __init__(self):
        self.spark = self._create_spark_session()
        self.spark.sparkContext.setLogLevel("ERROR")

    def run(self):
        # 从 Hive 表中读取数据
        df = self.spark.sql("SELECT * FROM checkin")

        # 显示原始数据
        df.show()

        # 去除包含任何空值的行并显示
        df_non_null = df.na.drop("any")
        df_non_null.show()

        # 统计原始数据的行数
        total_rows = df.count()
        print(f"原始数据的行数: {total_rows}")

        # 统计去重后的行数
        non_null_rows = df_non_null.count()
        print(f"去重后的行数: {non_null_rows}")
        # new_table_name = "clean_business"  # 新表的名称
        # df.write.option("path", "/user/hive/warehouse") \
        #      .format("hive") \
        #      .mode("overwrite") \
        #      .saveAsTable(new_table_name)
if __name__ == '__main__':
    TextRandJob().run()